package Market;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;

import Market.CBD.CBDMarketDataOuterClass.CBDMarketData;
import Market.CBD.DataSer;
import Market.CBD.ReadFromFile;;

public class DataProducer {

    public static void main(String[] argv) throws ParseException {
        int accelerated = 1000;
        System.out.println(Serdes.String().getClass());
        System.out.println(DataSer.class.getName());
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DataSer.class);
        KafkaProducer<String, CBDMarketData> producer = new KafkaProducer<>(props);

        ReadFromFile bondDataFormat = new ReadFromFile();
        bondDataFormat.readFromCsvFile("data/test_data.csv");
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd-HH:mm:ss.SSS");// 20220822-09:03:04.100
        long startTime = new Date().getTime();// ms
        // Date d = sdf.parse("20220822-09:03:04.100");
        // String s = sdf.format(d);
        long baseTime = sdf.parse(bondDataFormat.CBDMarkets.get(0).getTransactTime()).getTime();
        for (CBDMarketData CBDMarket : bondDataFormat.CBDMarkets) {
            long nowTime = new Date().getTime();
            long curTime = sdf.parse(CBDMarket.getTransactTime()).getTime();
            long diff = nowTime - startTime - (curTime - baseTime);
            if (diff < 0) {
                try {
                    System.out.printf("sleep %d s\n", 0-diff/accelerated);
                    Thread.sleep(0 - diff/accelerated);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

            producer.send(new ProducerRecord<>("stream-in","CBD", CBDMarket));
        }

        producer.close();
    }
}
